package com.weiyu.consumer;


import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Arrays;
import java.util.Properties;

/**
 * @author weiyu@gomeholdings.com
 * @description
 * @create 2017/6/29
 */
public class SimpleConsumer implements Runnable {
    private static final Logger logger = LoggerFactory.getLogger(SimpleConsumer.class);
    private static final String TOPIC = "test";
    private Consumer consumer;

    public SimpleConsumer(){
        Properties props = new Properties();
        props.put("bootstrap.servers","XXX");
        //group 代表一个消费组
        props.put("group.id", "test-consumer-group");
        props.put("enable.auto.commit", "true");
        props.put("auto.commit.interval.ms", "1000");
        props.put("session.timeout.ms", "30000");
        props.put("zookeeper.session.timeout.ms", "500");
        props.put("zookeeper.sync.time.ms", "250");
        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
//        props.put("zookeeper.connection.timeout.ms", "6000");
        consumer = new KafkaConsumer<>(props);
    }

    public void run() {
        consumer.subscribe(Arrays.asList(TOPIC));
        while (true){
            ConsumerRecords<String,String> records = consumer.poll(100);
            for (ConsumerRecord<String, String> record : records){
                System.out.println("offset = "+ record.offset()+", key = "+record.key()+", value = "+record.value());
                logger.info("offset = {}, key = {}, value = {}", record.offset(), record.key(), record.value());
            }
        }
    }

    public static void main(String[] args) {
       Thread t1 =  new Thread(new SimpleConsumer());
       t1.start();
    }

}
